package com.weipt.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

public class HelloWorkerFair {

    public static final String QUEUE_NAME = "hello-worker-fair—queue";
    public static final ExecutorService executorService = Executors.newFixedThreadPool(3);

    public static final  Channel finalChannel;

    static {
        try {
            finalChannel  = ConnectionUtils.getMqConn().createChannel(100); //todo 这个数据100.界面显示一个连接，括号100. 不太清楚和new channel的区别，待研究
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        consume1();
        consume2();
        publish("");
    }

    public static void publish(String msg) throws IOException, TimeoutException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel(); //创建一个新的管道
        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        for (int i = 0; i < 2000; i++) {
            channel.basicPublish("", QUEUE_NAME,  MessageProperties.PERSISTENT_TEXT_PLAIN, String.valueOf(i).getBytes());
        }
        //关闭连接
        channel.close();
        mqConn.close();
    }

    public static void consume1() throws IOException {
        finalChannel.queueDeclare(QUEUE_NAME, false,false,false,null);
        finalChannel.basicQos(100);
        //关闭自动应答，改成手动应答，按需分配，处理快的消费者提前应答获取，接受新的消费
        // 若干消费者，即使性能不一，但也是逐个分发，适用于消费者集群不稳定，配置不一样的情况
        //自动ack的话，相当于收到消息就应答，异步处理了，所以比较快
        finalChannel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(finalChannel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumeA："+new String(body));
                finalChannel.basicAck(envelope.getDeliveryTag(), false);
                executorService.execute(() -> doSomethingA(envelope.getDeliveryTag()));
            }
        });
    }

    public static void consume2() throws IOException {
        finalChannel.queueDeclare(QUEUE_NAME, false,false,false,null);
        //用于在消费者端设置预取（prefetch）的限制。预取是指消费者从队列中预取（prefetch）一定数量的消息，
        // 而不是一次只接收一个消息。通过设置预取限制，你可以控制消费者在一次确认之前能够接收多少消息。
        /**
         * prefetchSize（预取大小）： 一般情况下设为0，表示不限制预取的大小。
         * prefetchCount（预取数量）： 这是最重要的参数，它定义了消费者在确认之前可以预取的消息数量。如果设置为1，表示一次只预取一个消息；如果设置为0，表示不限制数量，即可以预取所有可用的消息。
         * global（全局设置）： 如果设置为true，表示这个预取限制将应用于整个通道，而不仅仅是当前消费者。如果设置为false（默认值），则表示仅对当前消费者应用预取限制。
         */
        //需要注意的是，预取的消息仅在消费者端的本地缓存中存储，而不是在RabbitMQ服务器上。
        finalChannel.basicQos(100); //当消息内容简单，消费者性能高的时候，可以增大这个值
        finalChannel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(finalChannel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumeB："+new String(body));
                finalChannel.basicAck(envelope.getDeliveryTag(), false);
                //此处放到dosomething中，就相当于处理完消息，再应答，性能提示不明显，比如之前处理500/s，用5个线程，也就是2500/s
                //但放到异步方法外，就说明收到消息就应答，即使处理失败也不管，此时消息会积压到java内存线程池的队列中。消息速率指数级提升
                //此时，ab方法的耗时，就无感，不影响。
                executorService.execute(() -> doSomethingB(envelope.getDeliveryTag()));
            }
        });
    }

    public static void doSomethingA(long tag){
        try {
            System.out.println("异步处理");
            Thread.sleep(50);
//            try {
//                finalChannel.basicAck(tag, false);
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void doSomethingB(long tag){
        try {
            System.out.println("异步处理");
            Thread.sleep(200);
//            try {
//                finalChannel.basicAck(tag, false);
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


}
